16 张图解开AbstractQueuedSynchronizer

（给ImportNew加星标，提高Java技能）

转自： 程序猿阿星

基础

AbstractQueuedSynchronizer 抽象同步队列简称 AQS，它是实现同步器的基础组件，如常用的 ReentrantLock、Semaphore、CountDownLatch 等。

AQS 定义了一套多线程访问共享资源的同步模板，解决了实现同步器时涉及的大量细节问题，能够极大地减少实现工作。

虽然大多数开发者可能永远不会使用 AQS 实现自己的同步器（JUC 包下提供的同步器基本足够应对日常开发），但是知道 AQS 的原理对于架构设计还是很有帮助的，面试还可以吹吹牛。

下面是 AQS 的组成结构：

AQS 由三部分组成：state 同步状态、Node 组成的 CLH 队列、ConditionObject 条件变量（包含 Node 组成的条件单向队列），下面会分别对这三部分做介绍。

先贴下 AbstractQueuedSynchronizer 提供的核心函数，混个脸熟就够了，后面会讲解。

状态

• getState()：返回同步状态；
• setState(int newState)：设置同步状态；
• compareAndSetState(int expect, int update)：使用 CAS 设置同步状态;
• isHeldExclusively()：当前线程是否持有资源。

独占资源（不响应线程中断）

• tryAcquire(int arg)：独占式获取资源，子类实现；
• acquire(int arg)：独占式获取资源模板；
• tryRelease(int arg)：独占式释放资源，子类实现；
• release(int arg)：独占式释放资源模板。

共享资源（不响应线程中断）

• tryAcquireShared(int arg)：共享式获取资源，返回值大于等于0则表示获取成功，否则获取失败，子类实现；
• acquireShared(int arg)：共享式获取资源模板；
• tryReleaseShared(int arg)：共享式释放资源，子类实现；
• releaseShared(int arg)：共享式释放资源模板。

这里补充下，获取独占、共享资源操作还提供超时与响应中断的扩展函数，有兴趣的读者可以去AbstractQueuedSynchronizer源码了解。

同步状态

在 AQS 中维护了一个同步状态变量 state，getState 函数获取同步状态，setState、compareAndSetState 函数修改同步状态。

对于 AQS来说，线程同步的关键是对 state 的操作。可以说获取、释放资源是否成功都是由 state 决定的。比如 state>0 代表可获取资源，否则无法获取。所以 state 的具体语义由实现者去定义。

现有的 ReentrantLock、ReentrantReadWriteLock、Semaphore、CountDownLatch 定义的 state 语义都不一样。

• ReentrantLock 的 state 用来表示是否有锁资源；
• ReentrantReadWriteLock 的 state 高 16 位代表读锁状态，低 16 位代表写锁状态；
• Semaphore 的 state 用来表示可用信号的个数；
• CountDownLatch 的 state 用来表示计数器的值。

CLH 队列

CLH 是 AQS 内部维护的 FIFO（先进先出）双端双向队列（方便尾部节点插入），基于链表数据结构。当一个线程竞争资源失败，就会将等待资源的线程封装成一个 Node 节点，通过 CAS 原子操作插入队列尾部。最终，不同的 Node 节点连接组成了一个 CLH 队列。所以说 AQS 通过 CLH 队列管理竞争资源的线程。

个人总结 CLH 队列具有如下几个优点：

• 先进先出保证了公平性；
• 非阻塞的队列，通过自旋锁和 CAS 保证节点插入和移除的原子性，实现无锁快速插入；
• 采用了自旋锁思想。所以 CLH 也是一种基于链表的可扩展、高性能、公平的自旋锁。

Node 内部类

Node 是 AQS 的内部类。每个等待资源的线程都会封装成 Node 节点组成 CLH 队列、等待队列。所以说 Node 是非常重要的部分，理解它是理解 AQS的第一步。

队列 Node 类中的变量都很好理解，只有 waitStatus、nextWaiter 没有细说，下面做个补充说明。

waitStatus 等待状态如下：

nextWaiter 特殊标记：

• Node 在 CLH 队列时，nextWaiter 表示共享式或独占式标记；
• Node 在条件队列时，nextWaiter 表示下个 Node 节点指针。

流程概述

线程获取资源失败，封装成 Node 节点从 CLH 队列尾部入队并阻塞线程。某线程释放资源时会把 CLH 队列首部 Node 节点关联的线程唤醒（此处的首部是指第二个节点，后面会细说）再次获取资源。

入队

获取资源失败的线程需要封装成 Node 节点，接着尾部入队。在 AQS 中提供 addWaiter 函数完成 Node 节点的创建与入队。

/**
 * @author: 程序猿阿星
 * @description:  Node节点入队-CLH队列
 * @param mode 标记  Node.EXCLUSIVE独占式 or Node.SHARED共享式
 */
private Node addWaiter(Node mode) {
    //根据当前线程创建节点，等待状态为0
    Node node = new Node(Thread.currentThread(), mode);
    // 获取尾节点
    Node pred = tail;
    if (pred != null) {
        //如果尾节点不等于null，把当前节点的前驱节点指向尾节点
        node.prev = pred;
        //通过cas把尾节点指向当前节点
        if (compareAndSetTail(pred, node)) {
            //之前尾节点的下个节点指向当前节点
            pred.next = node;
            return node;
        }
    }
    //如果添加失败或队列不存在，执行end函数
    enq(node);
    return node;
}

添加节点的时候，如果从 CLH队列已经存在，通过 CAS 快速将当前节点添加到队列尾部；如果添加失败或队列不存在，则指向 enq 函数自旋入队。

/**
 * @author: 程序猿阿星
 * @description: 自旋cas入队
 * @param node 节点
 */
private Node enq(final Node node) {
    for (;;) { //循环
        //获取尾节点
        Node t = tail;
        if (t == null) {
            //如果尾节点为空，创建哨兵节点，通过cas把头节点指向哨兵节点
            if (compareAndSetHead(new Node()))
                //cas成功，尾节点指向哨兵节点
                tail = head;
        } else {
            //当前节点的前驱节点设指向之前尾节点
            node.prev = t;
            //cas设置把尾节点指向当前节点
            if (compareAndSetTail(t, node)) {
                //cas成功，之前尾节点的下个节点指向当前节点
                t.next = node;
                return t;
            }
        }
    }
}

通过自旋 CAS 尝试往队列尾部插入节点，直到成功。自旋过程如果发现 CLH 队列不存在时，会初始化 CLH 队列。

入队过程流程如下图：

第一次循环：

1. 刚开始 CLH 队列不存在，head 与 tail 都指向 null；

2. 要初始化 CLH 队列，会创建一个哨兵节点。head 与 tail 都指向哨兵节点。

第二次循环

3.当前线程节点的前驱节点指向尾部节点（哨兵节点）；

4.设置当前线程节点为尾部，tail 指向当前线程节点；

5.前尾部节点的后驱节点指向当前线程节点（当前尾部节点）。

最后结合 addWaiter 与 enq 函数的入队流程图如下：

出队

CLH 队列中的节点都是获取资源失败的线程节点。当持有资源的线程释放资源时，会将 head.next 指向的线程节点唤醒（CLH 队列的第二个节点）。如果唤醒的线程节点获取资源成功，线程节点清空信息设置为头部节点（新哨兵节点），原头部节点出队（原哨兵节点）。

acquireQueued 函数中的部分代码：

//1.获取前驱节点
final Node p = node.predecessor();
//如果前驱节点是首节点，获取资源（子类实现）
if (p == head && tryAcquire(arg)) {
    //2.获取资源成功，设置当前节点为头节点，清空当前节点的信息，把当前节点变成哨兵节点
    setHead(node);
    //3.原来首节点下个节点指向为null
    p.next = null; // help GC
    //4.非异常状态，防止指向finally逻辑
    failed = false;
    //5.返回线程中断状态
    return interrupted;
}

private void setHead(Node node) {
    //节点设置为头部
    head = node;
    //清空线程
    node.thread = null;
    //清空前驱节点
    node.prev = null;
}

只需要关注1~3步骤即可，过程非常简单：假设获取资源成功，更换头部节点，并把头部节点的信息清除变成哨兵节点。

注意，这个过程是不需要使用 CAS来保证，因为只有一个线程能够成功获取到资源。

条件变量

Object 的 wait、notify 函数是配合 Synchronized 锁实现线程间同步协作的功能，AQS 的 ConditionObject 条件变量也提供这样的功能。通过 ConditionObject 的 await 和 signal 两类函数完成。

不同于 Synchronized 锁，一个 AQS 可以对应多个条件变量，而Synchronized只有一个。

如上图所示，ConditionObject 内部维护着一个单向条件队列。不同于 CHL 队列，条件队列只入队执行 await 的线程节点。并且加入条件队列的节点，不能在 CHL 队列， 条件队列出队的节点，会入队到 CHL 队列。

当某个线程执行了 ConditionObject 的 await 函数，阻塞当前线程。线程会被封装成 Node 节点添加到条件队列的末端，其他线程执行 ConditionObject 的 signal 函数，会将条件队列头部线程节点转移到 CHL 队列参与竞争资源，具体流程如下图：

最后补充下，条件队列Node类是使用nextWaiter变量指向下个节点，并且因为是单向队列，所以prev与next变量都是null

进阶

AQS 采用了模板方法设计模式，提供了两类模板：一类是独占式模板，另一类是共享形模式，对应的模板函数如下：

• 独占式：
• acquire 获取资源
• release 释放资源
• 共享式：
• acquireShared 获取资源
• releaseShared 释放资源

独占式获取资源

acquire 是个模板函数，模板流程就是线程获取共享资源。如果获取资源成功，线程直接返回；否则进入 CLH 队列，直到获取资源成功为止，且整个过程忽略中断的影响。

acquire 函数代码如下：

执行 tryAcquire 函数。tryAcquire 由子类实现，代表获取资源是否成功，如果资源获取失败，执行下面的逻辑：

• 执行 addWaiter 函数（前面已经介绍过）。根据当前线程创建出独占式节点，并入队 CLH 队列；
• 执行 acquireQueued 函数，自旋阻塞等待获取资源；
• 如果 acquireQueued 函数中获取资源成功，根据线程是否被中断状态，来决定执行线程中断逻辑。

acquire 函数的大致流程都清楚了，下面来分析下 acquireQueued 函数。线程封装成节点后，是如何自旋阻塞等待获取资源的，代码如下：

/**
 * @author: 程序猿阿星
 * @description: 自旋机制等待获取资源
 * @param node
 * @param arg
 * @return: boolean
 */
final boolean acquireQueued(final Node node, int arg) {
    //异常状态，默认是
    boolean failed = true;
    try {
        //该线程是否中断过，默认否
        boolean interrupted = false;
        for (;;) {//自旋
            //获取前驱节点
            final Node p = node.predecessor();
            //如果前驱节点是首节点，获取资源（子类实现）
            if (p == head && tryAcquire(arg)) {
                //获取资源成功，设置当前节点为头节点，清空当前节点的信息，把当前节点变成哨兵节点
                setHead(node);
                //原来首节点下个节点指向为null
                p.next = null; // help GC
                //非异常状态，防止指向finally逻辑
                failed = false;
                //返回线程中断状态
                return interrupted;
            }
            /**
             * 如果前驱节点不是首节点，先执行shouldParkAfterFailedAcquire函数，shouldParkAfterFailedAcquire做了三件事
             * 1.如果前驱节点的等待状态是SIGNAL，返回true，执行parkAndCheckInterrupt函数，返回false
             * 2.如果前驱节点的等大状态是CANCELLED，把CANCELLED节点全部移出队列（条件节点）
             * 3.以上两者都不符合，更新前驱节点的等待状态为SIGNAL，返回false
             * 
             * 使用LockSupport类的静态方法park挂起当前线程，直到被唤醒
             * 唤醒后检查当前线程是否被中断，返回该线程中断状态并重置中断状态
             */
            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {
                //该线程被中断过
                interrupted = true;
            }
        } finally {
            // 尝试获取资源失败并执行异常，取消请求，将当前节点从队列中移除
            if (failed) {
                cancelAcquire(node);
            }
        }
}

一图胜千言，核心流程图如下：

独占式释放资源

有获取资源，自然就少不了释放资源。AQS 中提供了 release 模板函数来释放资源，模板流程就是线程释放资源成功，唤醒 CLH队列的第二个线程节点（首节点的下个节点），代码如下：

/**
 * @author: 程序猿阿星
 * @description: 独占式-释放资源模板函数
 * @param arg
 * @return: boolean
 */
public final boolean release(int arg) {
    //释放资源成功，tryRelease子类实现
    if (tryRelease(arg)) {
        //获取头部线程节点
        Node h = head;
        //头部线程节点不为null，并且等待状态不为0
        if (h != null && h.waitStatus != 0) {
            //唤醒CHL队列第二个线程节点
            unparkSuccessor(h);
        }
        return true;
    }
    return false;
}


private void unparkSuccessor(Node node) {
    //获取节点等待状态
    int ws = node.waitStatus;
    if (ws < 0) {
        //cas更新节点状态为0
        compareAndSetWaitStatus(node, ws, 0);
    }

    //获取下个线程节点        
    Node s = node.next;
    //如果下个节点信息异常，从尾节点循环向前获取到正常的节点为止，正常情况不会执行
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev) {
            if (t.waitStatus <= 0) {
                s = t;
            }
        }
    }
    if (s != null) {
        //唤醒线程节点
        LockSupport.unpark(s.thread);
    }
}

release 逻辑非常简单，流程图如下：

共享式获取资源

acquireShared 是个模板函数，模板流程就是线程获取共享资源。如果获取到资源，线程直接返回；否则进入CLH队列，直到获取到资源为止，且整个过程忽略中断的影响。acquireShared 函数代码如下：

/**
 * @author: 程序猿阿星
 * @description: 共享式-获取资源模板函数
 * @param arg
 * @return: void
 */
public final void acquireShared(int arg) {
    /**
     * 1.负数表示失败
     * 2.0表示成功，但没有剩余可用资源
     * 3.正数表示成功且有剩余资源
     */
    if (tryAcquireShared(arg) < 0) //获取资源失败，tryAcquireShared子类实现
        //自旋阻塞等待获取资源
        doAcquireShared(arg);
}

doAcquireShared 函数与独占式的 acquireQueued 函数逻辑基本一致，唯一的区别就是下图红框部分：

• 节点的标记是共享式；
• 获取资源成功，还会唤醒后续资源。因为资源数可能>0，代表还有资源可获取，所以需要做后续线程节点的唤醒。

共享式释放资源

AQS 中提供了 releaseShared 模板函数来释放资源，模板流程就是线程释放资源成功，唤醒 CHL 队列的第二个线程节点（首节点的下个节点），代码如下：

/**
 * @author: 程序猿阿星
 * @description: 共享式-释放资源模板函数
 * @param arg
 * @return: boolean
 */
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {//释放资源成功，tryReleaseShared子类实现
        //唤醒后继节点
        doReleaseShared();
        return true;
    }
    return false;
}

private void doReleaseShared() {
    for (;;) {
        //获取头节点
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;

            if (ws == Node.SIGNAL) {//如果头节点等待状态为SIGNAL
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))//更新头节点等待状态为0
                    continue;            // loop to recheck cases
                //唤醒头节点下个线程节点
                unparkSuccessor(h);
            }
            //如果后继节点暂时不需要被唤醒，更新头节点等待状态为PROPAGATE
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;               
        }
        if (h == head)              
            break;
    }
}

与独占式释放资源区别不大，都是唤醒头节点的下个节点，就不做过多描述了。

实战

说了这么多理论，现在到实战环节了。正如前文所述，AQS 定义了一套多线程访问共享资源的同步模板，解决了实现同步器时涉及的大量细节问题，能够极大地减少实现工作。现在我们基于 AQS 实现一个不可重入的独占锁，直接使用 AQS 提供的独占式模板，只需明确 state 的语义与实现 tryAcquire 与 tryRelease 函数（获取资源与释放资源），在这里 state 为 0 表示锁没有被线程持有，state 为 1 表示锁已经被某个线程持有。由于是不可重入锁，所以不需要记录持有锁线程的获取锁次数。

不可重入的独占锁代码如下：

/**
 * @Author 程序猿阿星
 * @Description 不可重入的独占锁
 */
public class NonReentrantLock implements Lock {
    /**
     * @Author 程序猿阿星
     * @Description 自定义同步器
     */
    private static class Sync extends AbstractQueuedSynchronizer {
        /**
         * 锁是否被线程持有
         */
        @Override
        protected boolean isHeldExclusively() {
            //0：未持有 1：已持有
            return super.getState() == 1;
        }

        /**
         * 获取锁
         */
        @Override
        protected boolean tryAcquire(int arg) {
            if (arg != 1) {
                //获取锁操作，是需要把state更新为1，所以arg必须是1
                throw new RuntimeException("arg not is 1");
            }
            if (compareAndSetState(0, arg)) {//cas 更新state为1成功，代表获取锁成功
                //设置持有锁线程
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        /**
         * 释放锁
         */
        @Override
        protected boolean tryRelease(int arg) {
            if (arg != 0) {
                //释放锁操作，是需要把state更新为0，所以arg必须是0
                throw new RuntimeException("arg not is 0");
            }
            //清空持有锁线程
            setExclusiveOwnerThread(null);
            //设置state状态为0，此处不用cas，因为只有获取锁成功的线程才会执行该函数，不需要考虑线程安全问题
            setState(arg);
            return true;
        }

        /**
         * 提供创建条件变量入口
         */
        public ConditionObject createConditionObject() {
            return new ConditionObject();
        }
    }

    private final Sync sync = new Sync();

    /**
     * 获取锁
     */
    @Override
    public void lock() {
        //Aqs独占式-获取资源模板函数
        sync.acquire(1);
    }
        
    /**
     * 获取锁-响应中断
     */
    @Override
    public void lockInterruptibly() throws InterruptedException {
        //Aqs独占式-获取资源模板函数(响应线程中断)
        sync.acquireInterruptibly(1);
    }

    /**
     * 获取锁是否成功-不阻塞
     */
    @Override
    public boolean tryLock() {
        //子类实现
        return sync.tryAcquire(1);
    }
    
    /**
     * 获取锁-超时机制
     */
    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        //Aqs独占式-获取资源模板函数(超时机制)
        return sync.tryAcquireNanos(1,unit.toNanos(time));
    }
    
    /**
     * 释放锁
     */
    @Override
    public void unlock() {
        //Aqs独占式-释放资源模板函数
        sync.release(0);
    }
    
    /**
     * 创建条件变量
     */
    @Override
    public Condition newCondition() {
        return sync.createConditionObject();
    }
}

NonReentrantLock 定义了一个内部类 Sync，Sync 用来实现具体的锁操作，它继承了 AQS，因为使用的是独占式模板，所以重写 tryAcquire 与 tryRelease 函数，另外提供了一个创建条件变量的入口，下面使用自定义的独占锁来同步两个线程对 j++。

private static int j = 0;

public static void main(String[] agrs) throws InterruptedException {
    NonReentrantLock  nonReentrantLock = new NonReentrantLock();

    Runnable runnable = () -> {
        //获取锁
        nonReentrantLock.lock();
        for (int i = 0; i < 100000; i++) {
            j++;
        }
        //释放锁
        nonReentrantLock.unlock();
    };

    Thread thread = new Thread(runnable);
    Thread threadTwo = new Thread(runnable);

    thread.start();
    threadTwo.start();

    thread.join();
    threadTwo.join();

    System.out.println(j);
}

无论执行多少次输出内容都是：200000。

AQS简化流程图

- EOF -

推荐阅读  点击标题可跳转

看完本文有收获？请转发分享给更多人

关注「ImportNew」，提升Java技能

点赞和在看就是最大的支持❤️